package com.tang.lts.core.biz.logger;

import com.tang.lts.core.biz.logger.domain.JobLogPo;
import com.tang.lts.core.cluster.Config;
import com.tang.lts.core.constant.ExtConfig;
import com.tang.lts.core.factory.NameThreadFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author Tang
 * @classname LazyJobLogger
 * @description [  ]
 * @date 2022/5/8 21:51
 */
public class LazyJobLogger implements JobLogger {

    /**
     * 无界queue
     */
    private BlockingQueue<JobLogPo> memoryQueue = new LinkedBlockingDeque<>();
    private int batchFlushSize;
    private int overflowSize;
    private int maxMemoryLogSize;

    private JobLogger delegate;

    private AtomicBoolean flushing = new AtomicBoolean(false);

    public LazyJobLogger(Config config, JobLogger delegate) {
        this.delegate = delegate;
        // 最大内存日志
        maxMemoryLogSize = config.getParameter(ExtConfig.LAZY_JOB_LOGGER_MEM_SIZE, 1000);
        // 扫描频率
        int flushPeriod = config.getParameter(ExtConfig.LAZY_JOB_LOGGER_CHECK_PERIOD, 3);
        // 批量刷新阈值
        batchFlushSize = config.getParameter(ExtConfig.LAZY_JOB_LOGGER_BATCH_FLUSH_SIZE, 100);
        // 超出值
        overflowSize = config.getParameter(ExtConfig.LAZY_JOB_LOGGER_OVERFLOW_SIZE, 10000);

        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
                new NameThreadFactory("LazyJobLogger", true));
        ScheduledFuture<?> scheduledFuture = executor.scheduleWithFixedDelay(() -> {
            if (flushing.compareAndSet(false, true)) {
                checkAndFlush();
            }
        }, flushPeriod, flushPeriod, TimeUnit.SECONDS);

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            checkAndFlush();
            scheduledFuture.cancel(true);
        }));

    }

    private void checkAndFlush() {
        try {
            int nowSize = memoryQueue.size();
            if (nowSize == 0) {
                return;
            }
            ArrayList<JobLogPo> batch = new ArrayList<>();
            for (int i = 0; i < nowSize; i++) {
                JobLogPo poll = memoryQueue.poll();
                batch.add(poll);

                if (batch.size() >= batchFlushSize) {
                    flush(batch);
                }
            }
            if (batch.size() > 0) {
                flush(batch);
            }
        } finally {
            flushing.compareAndSet(true, false);
        }
    }

    private void flush(ArrayList<JobLogPo> batch) {
        boolean flushSuccess = false;
        try {
            this.delegate.log(batch);
            flushSuccess = true;
        } finally {
            if (!flushSuccess) {
                memoryQueue.addAll(batch);
            }
            batch.clear();
        }
    }

    public void checkOverflowSize() {
        if (memoryQueue.size() > overflowSize) {
            throw new RuntimeException("memory log size is" + memoryQueue.size());
        }
    }

    public void checkCapacity() {
        if (memoryQueue.size() > maxMemoryLogSize) {
            if (flushing.compareAndSet(false, true)) {
                // 异步刷新
                new Thread(() -> {
                    checkAndFlush();
                }).start();
            }
        }
    }

    @Override
    public void log(JobLogPo jobLogPo) {
        checkOverflowSize();
        memoryQueue.offer(jobLogPo);
        checkCapacity();
    }

    @Override
    public void log(List<JobLogPo> jobLogPos) {
        checkOverflowSize();
        for (JobLogPo jobLogPo : jobLogPos) {
            log(jobLogPo);
        }
        checkCapacity();
    }
}